package com.sand.pro.flinksql_pro.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.concurrent.ExecutionException;

public class Kfk2Mysql {
    public static final String DB_URL = "jdbc:derby:memory:upsert";
    public static final String USER_TABLE = "USER_TABLE";
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 注册SourceTable: source_sensor
        //latest-offset  earliest-offset
        tableEnv.executeSql("create table source_sensor (id string, ts bigint, vc int) with("
                + "'connector' = 'kafka',"
                + "'topic' = 'topic_source_sensor',"
                + "'properties.bootstrap.servers' = '172.28.246.66:9092,172.28.246.67:9092,172.28.246.68:9092',"
                + "'properties.group.id' = 'sqlid',"
                + "'scan.startup.mode' = 'earliest-offset',"
                + "'format' = 'json'"
                + ")");
        tableEnv.executeSql(
                "CREATE TABLE user_sink (id string, ts bigint, vc int) WITH (\n"
                        + "  'connector' = 'jdbc',"
                        + "  'url'='"
                        + DB_URL
                        + "',"
                        + "  'table-name' = '"
                        + USER_TABLE
                        + "',"
                        + "  'sink.buffer-flush.max-rows' = '2',"
                        + "  'sink.buffer-flush.interval' = '0'"
                        + // disable async flush
                        ")");
        try {
            tableEnv.executeSql("INSERT INTO user_sink SELECT * FROM source_sensor").await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }
}
